package top.scauwlt.remoting.transport.codec;

import io.netty.buffer.ByteBuf;
import io.netty.channel.ChannelHandlerContext;
import io.netty.handler.codec.MessageToByteEncoder;
import lombok.extern.slf4j.Slf4j;
import top.scauwlt.compress.Compress;
import top.scauwlt.enums.CompressTypeEnum;
import top.scauwlt.enums.SerializationTypeEnum;
import top.scauwlt.extension.ExtensionLoader;
import top.scauwlt.remoting.constants.RpcConstants;
import top.scauwlt.remoting.dto.RpcMessage;
import top.scauwlt.serialize.Serializer;

import java.util.concurrent.atomic.AtomicInteger;

/**
 * @Author wmm
 * @Date 2022/7/25 22:41
 */
@Slf4j
public class RpcMessageEncoder extends MessageToByteEncoder<RpcMessage> {

    private static final AtomicInteger ATOMIC_INTEGER = new AtomicInteger(0);

    @Override
    protected void encode(ChannelHandlerContext ctx, RpcMessage rpcMessage, ByteBuf out) {
        try {
            out.writeBytes(RpcConstants.MAGIC_CODE);
            out.writeByte(RpcConstants.VERSION);
            // 这里先跳过四个字节，留着放消息总长度，主要是提前算长度会让代码变乱
            out.writerIndex(out.writerIndex() + 4);
            //消息类型
            byte messageType = rpcMessage.getMessageType();
            out.writeByte(messageType);
            //消息序列化方式
            out.writeByte(rpcMessage.getCodec());
            //消息压缩方式
            out.writeByte(rpcMessage.getCompress());
            //消息请求Id
            out.writeInt(ATOMIC_INTEGER.getAndIncrement());

            byte[] bodyBytes = null;
            if (messageType != RpcConstants.HEARTBEAT_REQUEST_TYPE && messageType != RpcConstants.HEARTBEAT_RESPONSE_TYPE) {
                //序列化对象
                String codecName = SerializationTypeEnum.getName(rpcMessage.getCodec());
                Serializer serializer = ExtensionLoader.getExtensionLoader(Serializer.class).getExtension(codecName);
                bodyBytes = serializer.serialize(rpcMessage.getData());

                //压缩数据
                String compressName = CompressTypeEnum.getName(rpcMessage.getCompress());
                Compress compress = ExtensionLoader.getExtensionLoader(Compress.class).getExtension(compressName);
                bodyBytes = compress.compress(bodyBytes);
            }

            // 写入数据体
            int fullLength = RpcConstants.HEAD_LENGTH;
            if (bodyBytes != null) {
                out.writeBytes(bodyBytes);
                fullLength += bodyBytes.length;
            }

            // 修改写指针，写回消息长度
            int writerIndex = out.writerIndex();
            out.writerIndex(RpcConstants.MAGIC_CODE.length + 1);
            out.writeInt(fullLength);
            out.writerIndex(writerIndex);

        } catch (Exception e) {
            log.error("encode message to RpcMessage fail: [{}]", e.getMessage());
        }
    }
}
